#include "rnet.h"
#include "main_service.h"
#include "read_task.h"

main_service::main_service()
{
}	

main_service::~main_service( )
{
	for ( map<int, rtask*>::iterator it = mm_task.begin() ; it != mm_task.end() ; it++ )
	{
		rtask* p_task = it->second ;
		rmutex(p_task->m_lock);
		delete_task(*p_task);
		delete p_task ;
		p_task = NULL ;
	}

	rlog << "main_service quit ..." << endl;
}

bool main_service::start(void)
{
	if ( !mservice_epoll.create(gsh_listen_port) )
	{
		rlog << "mservice_epoll.create() error" << endl ;
		return false ;
	}

	pthread_t tid_epoll_wait ;
	if ( pthread_create(&tid_epoll_wait, NULL, epoll_wait_routine, this) != 0 )
	{
		rlog << "pthread_create() : epoll_wait_routine : " << strerror( errno ) << endl;
		return false;
	}	

	pthread_t tid_epoll_accept ;
	if ( pthread_create(&tid_epoll_accept, NULL, epoll_accept_routine, this) != 0 )
	{
		rlog << "pthread_create() : epoll_accept_routine : " << strerror( errno ) << endl;
		return false;
	}

	return true ;
}

void main_service::init_task( const int i_socket )
{
	map<int, rtask*>::iterator it = mm_task.find(i_socket) ;
	if ( it == mm_task.end() )
		mm_task[i_socket] = new rtask();

	rtask* p_task = mm_task.find(i_socket)->second ;
	rmutex(p_task->m_lock) ;
	p_task->init(i_socket);
	mservice_epoll.event_add(i_socket) ;
}

void main_service::delete_task(rtask& r_task)
{
	mservice_epoll.event_del(r_task.mi_fd);
	r_task.clear();
}

void main_service::clear_task( const int i_socket )
{
	if ( i_socket == -1 )
		return ;
		
	map<int, rtask*>::iterator it = mm_task.find(i_socket) ;
	if ( it == mm_task.end() )
		return ;
		
	rtask* p_task = it->second ;	
	rmutex(p_task->m_lock) ;
	delete_task(*p_task);
}

void* main_service::epoll_accept_routine(void* arg)
{
	pthread_detach(pthread_self());

	if ( arg == NULL )
	{
		rlog << "argument = NULL" << endl ; 
		return NULL ;
	}
	
	main_service* p_this = reinterpret_cast<main_service*>( arg );
	p_this->epoll_accept_entry();

	return NULL;
}

void main_service::epoll_accept_entry(void)
{
	while( true )
	{
		int i_socket = mservice_epoll.accept() ;
		if ( i_socket < 0 )
		{
			rlog << "mservice_epoll.accept()" << endl ;
			continue ;
		}

		init_task(i_socket) ;
	}
}

void* main_service::epoll_wait_routine(void* arg)
{
	pthread_detach(pthread_self());

	if ( arg == NULL )
	{
		rlog << "argument = NULL" << endl ; 
		return NULL ;
	}
	
	main_service* p_this = reinterpret_cast<main_service*>( arg );
	p_this->epoll_wait_entry();

	return NULL;
}

void main_service::epoll_wait_entry(void)
{
	while(true) 
	{
		int nfds = mservice_epoll.wait() ;

		if(nfds == 0) 
			continue;

		if(nfds < 0)
		{
			rlog << "mservice_epoll.wait() error" << endl;
			continue ;
		}

		for(int i = 0; i < nfds; ++i)
		{
			int i_socket = mservice_epoll.mresult_event[i].data.fd ;

			rlog << "main-task(" << i_socket << ") event" << endl ;

			if ( mservice_epoll.mresult_event[i].events & EPOLLERR )
			{
				rlog << "main-task(" << i_socket << ") error event" << endl ;

				clear_task(i_socket);
				continue ;
			}
			
			rthreadpool::_instance().add_task(new read_task(*mm_task[i_socket]));
		}
	}
}

